Apache Storm 是一个分布式实时计算系统，用于处理大规模数据流。它支持 Java 编程，以下是一个简单的 Storm 示例，展示如何编写一个 Java 程序来处理数据流。

---

### **1. 环境准备**
确保你已经安装了以下工具：
- JDK 8+
- Apache Storm
- Maven

---

### **2. 创建 Maven 项目**
在 `pom.xml` 中添加 Storm 依赖：
```xml
<dependencies>
    <dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-core</artifactId>
        <version>2.4.0</version>
    </dependency>
</dependencies>
```

---

### **3. 编写 Storm 拓扑**
Storm 的核心是 **Topology**，它由 **Spout**（数据源）和 **Bolt**（数据处理单元）组成。

#### **3.1 Spout 示例**
Spout 是数据源，负责从外部系统（如 Kafka、文件等）读取数据并发送到 Bolt。
```java
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class SimpleSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;

    @Override
    public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        // 模拟数据源，发送一个字符串
        collector.emit(new Values("Hello, Storm!"));
        try {
            Thread.sleep(1000); // 每秒发送一次
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }
}
```

#### **3.2 Bolt 示例**
Bolt 是数据处理单元，负责处理从 Spout 接收的数据。
```java
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

public class SimpleBolt extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        // 处理数据
        String word = input.getStringByField("word");
        System.out.println("Received: " + word);

        // 确认处理完成
        collector.ack(input);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 不需要输出字段
    }
}
```

#### **3.3 Topology 示例**
将 Spout 和 Bolt 组合成一个拓扑。
```java
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;

public class SimpleTopology {
    public static void main(String[] args) {
        // 创建拓扑构建器
        TopologyBuilder builder = new TopologyBuilder();

        // 设置 Spout
        builder.setSpout("SimpleSpout", new SimpleSpout());

        // 设置 Bolt
        builder.setBolt("SimpleBolt", new SimpleBolt()).shuffleGrouping("SimpleSpout");

        // 配置
        Config config = new Config();
        config.setDebug(true);

        // 本地模式运行
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("SimpleTopology", config, builder.createTopology());

        // 运行 10 秒后停止
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 关闭集群
        cluster.shutdown();
    }
}
```

---

### **4. 运行代码**
1. 编译并运行程序：
   ```bash
   mvn clean package
   java -cp target/storm-example-1.0.jar SimpleTopology
   ```
2. 观察控制台输出，每秒钟会打印一次 `Received: Hello, Storm!`。

---

### **5. 代码说明**
- **Spout**：数据源，负责生成数据并发送到 Bolt。
- **Bolt**：数据处理单元，负责处理从 Spout 接收的数据。
- **Topology**：将 Spout 和 Bolt 组合成一个数据处理流程。
- **LocalCluster**：在本地模式下运行 Storm 拓扑。

---

### **6. 扩展**
- **多 Bolt**：可以添加多个 Bolt，形成复杂的数据处理流程。
- **分组策略**：使用 `shuffleGrouping`、`fieldsGrouping` 等分组策略控制数据分发。
- **分布式模式**：将拓扑提交到 Storm 集群运行。

---

### **7. 分布式模式运行**
1. 将拓扑打包为 JAR 文件：
   ```bash
   mvn clean package
   ```
2. 提交到 Storm 集群：
   ```bash
   storm jar target/storm-example-1.0.jar SimpleTopology
   ```

---

### **8. 注意事项**
- 在分布式模式下，确保 Storm 集群已启动。
- 如果需要从外部系统（如 Kafka、数据库）读取数据，可以在 Spout 中实现相应的逻辑。

---

希望这个示例对你有帮助！如果有其他问题，请随时提问。